package com.fwmagic.spark.core

import org.apache.spark.sql.{Dataset, SparkSession}

/**
  * 基础操作
  *
  * 持久化：cache、persist
  * 创建临时视图：createTempView、createOrReplaceTempView
  * 获取执行计划：explain
  * 查看schema：printSchema
  * 写数据到外部存储：write
  * dataset与dataframe互相转换：as、toDF
  *
  */
object BasicOperation {

  def main(args: Array[String]): Unit = {
    val spark:SparkSession = SparkSession.builder()
      .appName("BasicOperation")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val employee = spark.read.json("/Users/fangwei/learn/mycode/workspace/fwmagic-spark/src/main/resources/employee.json")

    //持久化：如果对一个dataset重复计算两次的话，可以先对这个dateframe/dateframe/da化后再进行操作
    employee.cache()

    println(employee.count())

    println(employee.count())

    //创建临时视图是为了可以对数据使用sql语句
    employee.createOrReplaceTempView("t_employee")

    spark.sql("select * from t_employee").show()

    println("====================")

    spark.sql("select * from t_employee where age >40").show()

    //获取spark sql的执行计划
    /**
      * dataframe/dataset。比如执行了一个sql语句获取dataframe，实际上内部包含了一个logical plan,逻辑执行计划，
      * 设计执行的时候，首先会通过底层的catalyst optimizer，生成物理执行计划，比如说会做一些优化，如：push,filter
      * 还会通过whole-stage code generation技术去自动化执行代码，提升性能优化
      *
      */
    spark.sql("select * from t_employee").explain()

    //打印schema
    employee.printSchema()

    //查出数据写到hdfs中，以local为例
    //spark.sql("select * from t_employee where age >40").write.json("/Users/fangwei/learn/mycode/workspace/fwmagic-spark/src/main/resources/emp40")

    //DataFrame to DataSet
    val empDS: Dataset[Emplpyee] = employee.as[Emplpyee]

    empDS.show()

    empDS.printSchema()

    //DataSet to DataFrame
    val dataFrame = empDS.toDF()

    spark.stop()
  }

  case class Emplpyee(name:String,age:Long,depId:Long,gender:String,salary:Long)

}
